package com.example.push.listener;

import com.example.push.domain.MessageRecord;
import com.example.push.repository.MessageRecordRepository;
import lombok.RequiredArgsConstructor;
import lombok.extern.slf4j.Slf4j;
import org.springframework.kafka.annotation.KafkaListener;
import org.springframework.stereotype.Component;

import java.time.LocalDateTime;

@Slf4j
@Component
@RequiredArgsConstructor
public class MessageListener {

    private final MessageRecordRepository messageRecordRepository;

    @KafkaListener(topics = "push-topic", groupId = "${spring.kafka.consumer.group-id}")
    public void listen(String message) {
        log.info("Received message: {}", message);

        try {
            // 创建消息记录
            MessageRecord record = new MessageRecord(generateMessageId(), message);
            messageRecordRepository.save(record);

            // 模拟消息处理
            processMessage(record);

            // 更新处理状态
            record.setStatus("PROCESSED");
            record.setProcessedAt(LocalDateTime.now());
            messageRecordRepository.save(record);

            log.info("Message processed successfully: {}", record.getMessageId());
        } catch (Exception e) {
            log.error("Error processing message: {}", message, e);
        }
    }

    private void processMessage(MessageRecord record) {
        // 模拟消息处理耗时
        try {
            Thread.sleep(100);
        } catch (InterruptedException e) {
            Thread.currentThread().interrupt();
            throw new RuntimeException("Message processing interrupted", e);
        }
    }

    private String generateMessageId() {
        return java.util.UUID.randomUUID().toString();
    }
}